package cn.doitedu.rtmk.demo7;

import cn.doitedu.rtmk.common.EventBean;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.util.Collector;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.roaringbitmap.longlong.Roaring64Bitmap;

import java.io.IOException;

/**
 * 规则id: rule-1：
 * 静态画像条件： age:[20,39] , gender = male
 * 实时动态画像标签条件: w行为次数 > 3
 * 触发条件: u行为
 */

/**
 * 规则id: rule-2：
 * 静态画像条件： age:[25,30] , active_level>2
 * 实时动态画像标签条件: g行为次数 >=4
 * 触发条件: p行为
 */

public class RuleModel_1_Calculator implements RuleCalculator {

    JSONObject jsonObject;
    ValueState<Integer> rule1State;

    JSONObject paramJsonObject;
    String ruleId;
    Roaring64Bitmap preSelectUsersBitmap;


    @Override
    public void init(RuntimeContext runtimeContext, String ruleParamJsonStr, Roaring64Bitmap preSelectUsersBitmap) throws IOException {

        // 外面注入的 规则的 静态画像条件  预圈选人群
        this.preSelectUsersBitmap = preSelectUsersBitmap;

        jsonObject = new JSONObject();

        // 申请一个keyedState，来记录每个用户的规则1的w行为次数
        rule1State = runtimeContext.getState(new ValueStateDescriptor<Integer>("rule-1-state", Integer.class));

        // 将规则参数，解析成fastjson内部的通用的对象
        paramJsonObject = JSON.parseObject(ruleParamJsonStr);
        ruleId = paramJsonObject.getString("rule_id");


    }

    @Override
    public void calc(EventBean eventBean, Collector<String> collector) throws IOException {

        String eventId = eventBean.getEventId();

        // 先判断该行为事件的行为人，如果不属于本规则的目标人群，则直接返回，啥也不做
        if(!preSelectUsersBitmap.contains(eventBean.getUid())) return;


        // 0. 判断事件是否是动态画像的目标事件，如果是，则做动态画像统计
        JSONObject realtimeProfileObject = paramJsonObject.getJSONObject("realtime_profile");
        String realtimeProfileEventId = realtimeProfileObject.getString("eventId");
        if (eventId.equals(realtimeProfileEventId)) {
            rule1State.update( ( rule1State.value() == null ? 0 : rule1State.value()) + 1);
        }

        // 1.判断事件是否是触发条件
        String fireActionEventId = paramJsonObject.getString("fire_action");
        if (eventId.equals(fireActionEventId)) {
            // 2.判断该用户的 “动态画像条件”是否已满足
            Integer realtimeEventCnt = realtimeProfileObject.getInteger("eventCnt");
            if (rule1State.value() != null && rule1State.value() >= realtimeEventCnt) {

                // 3. 如果上面的比较全部通过，则说明该用户已经命中该规则
                jsonObject.put("uid", eventBean.getUid());
                jsonObject.put("timestamp", eventBean.getTimestamp());
                jsonObject.put("rule_id", ruleId);

                collector.collect(jsonObject.toJSONString());

            }
        }
    }
}
